package com.xiaofeng.springboot.kafka;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.junit.Test;

/**
 * 消费者
 * @author chenzf
 * @date 2017年6月9日上午10:13:20
 */
@SuppressWarnings({ "static-method", "resource" })
public class KafkaTestConsumer {

	public void consume() {
		Properties props = new Properties();
		props.put("bootstrap.servers", "localhost:9092");
		props.put("group.id", "test");
		props.put("enable.auto.commit", "true"); //自动commit
		props.put("auto.commit.interval.ms", "1000"); //定时commit的周期
		props.put("session.timeout.ms", "30000"); //consumer活性超时时间
		props.put("max.poll.records", "20");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
		consumer.subscribe(Arrays.asList(KafkaTopic.TEST_TOPICE)); //subscribe，foo，bar，两个topic
		//while (true) {
		ConsumerRecords<String, String> records = consumer.poll(100); //100是超时等待时间
		records.forEach(record -> {
			System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
		});
		//}
	}

	@Test
	public void receive() {
		consume();
	}
}
